package com.gitee.andy.rpc.server;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import com.gitee.andy.rpc.core.MessageDecoder;
import com.gitee.andy.rpc.core.MessageEncoder;
import com.gitee.andy.rpc.entity.RPC;
import com.gitee.andy.rpc.entity.RpcMethodId;
import com.gitee.andy.rpc.entity.RpcRequest;
import com.gitee.andy.util.Clazzs;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultThreadFactory;

public class RpcServer implements ApplicationContextAware, InitializingBean {

	private static final Logger log = LoggerFactory.getLogger(RpcServer.class);

	private EventLoopGroup bossGroup;
	private EventLoopGroup workerGroup;
	private ServerBootstrap serverBootstrap;

	private ExecutorService es;

	private String pkgName;
	private int inetPort;
	private static Map<RpcMethodId, RpcMethod> methodInvokes = new HashMap<>();

	private RpcServer() {
		es = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2,
				new DefaultThreadFactory("RPC-METHOD-HANDLER"));
	}

	public void execute(Runnable task) {
		es.execute(task);
	}

	public void shutdown() {
		bossGroup.shutdownGracefully();
		workerGroup.shutdownGracefully();
		es.shutdown();
	}

	public RpcMethod get(RpcMethodId rpcMethodId) {
		return methodInvokes.get(rpcMethodId);
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		bossGroup = new NioEventLoopGroup(1);
		workerGroup = new NioEventLoopGroup();
		serverBootstrap = new ServerBootstrap();
		serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
				.option(ChannelOption.SO_BACKLOG, 100).handler(new LoggingHandler(LogLevel.INFO))
				.childHandler(new ChannelInitializer<SocketChannel>() {
					@Override
					public void initChannel(SocketChannel ch) throws Exception {
						ch.pipeline().addLast(new MessageDecoder(RpcRequest.class), new MessageEncoder(),
								new RpcServerHandler(RpcServer.this));
					}
				});
		serverBootstrap.bind(inetPort).sync();
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		Clazzs.getClasses(pkgName).stream().filter(clazz -> clazz.isAnnotationPresent(RPC.class) && clazz.isInterface())
				.forEach(clazz -> {
					Object bean = applicationContext.getBean(clazz);
					Arrays.stream(clazz.getMethods()).forEach(method -> {
						RpcMethodId rpcMethodId = new RpcMethodId(method.getDeclaringClass().getName(),
								method.getName(), method.getParameterTypes());
						methodInvokes.put(rpcMethodId, new RpcMethod(bean, method));
					});
					log.info("scan clazz>>>{}", clazz.getName());
				});
	}

	public String getPkgName() {
		return pkgName;
	}

	public void setPkgName(String pkgName) {
		this.pkgName = pkgName;
	}

	public int getInetPort() {
		return inetPort;
	}

	public void setInetPort(int inetPort) {
		this.inetPort = inetPort;
	}

}
